package com.tarena.rabbitmq.test;

import com.rabbitmq.client.*;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * 所有和死信队列有关的测试代码全都写到这里
 */
public class DemoDeadLetter {
    /*
        无论是声明组件,绑定关系,发送消息,消费消息
        都需要连接对象
     */
    private Channel channel;
    @Before
    public void init() throws IOException, TimeoutException {
        //创建连接工厂(有工厂类 就说明这个链接 不可能是new之后set set就搞定的)
        ConnectionFactory factory=new ConnectionFactory();
        //对连接工厂设置蓝图 如何连接 连接谁 localhost:5672 guest guest用户 访问虚拟主机 /
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setUsername("guest");
        factory.setPassword("guest");
        factory.setVirtualHost("/");
        //从工厂拿到长连接
        channel=factory.newConnection().createChannel();
    }
    //准备一下四个组件的名称
    private static final String B_EX="biz_ex";//业务交换机
    private static final String B_QUEUE="biz_queue";//业务队列
    private static final String DL_EX="dl_ex";//死信交换机
    private static final String DL_QUEUE="dl_queue";//死信队列
    //业务交换机 绑定业务队列路由
    private static final String B_ROUTING_KEY="biz_routing_key";
    //死信队列和死信交换机 绑定路由
    private static final String DL_ROUTING_KEY="dl_routing_key";

    @Test
    public void componentsDeclare() throws IOException {
        //业务交换机 不持久化 不自动删除 没有额外args属性
        channel.exchangeDeclare(B_EX,"direct",false,false,null);
        //死信交换机 不持久化 不自动删除 没有额外args属性
        channel.exchangeDeclare(DL_EX,"direct",false,false,null);
        //声明B_QUEUE队列
        Map<String,Object> args=new HashMap<>();
        args.put("x-dead-letter-exchange",DL_EX);
        args.put("x-dead-letter-routing-key",DL_ROUTING_KEY);
        args.put("x-message-ttl",30000);
        channel.queueDeclare(B_QUEUE,false,false,false,args);
        //声明DL_QUEUE队列
        channel.queueDeclare(DL_QUEUE,false,false,false,null);

        //业务一对绑定
        channel.queueBind(B_QUEUE,B_EX,B_ROUTING_KEY);
        //死信一对绑定
        channel.queueBind(DL_QUEUE,DL_EX,DL_ROUTING_KEY);
    }
    //测试消息生产者发送
    @Test
    public void sendDeadMessage() throws IOException {
        String msg="我是一条死信消息";
        channel.basicPublish(B_EX,B_ROUTING_KEY,null,msg.getBytes());
    }
    //测试发送一条延迟消息
    @Test
    public void sendDelayMessage() throws IOException {
        String msg="我是一条延迟消息";
        LocalTime localTime = LocalDateTime.now().toLocalTime();
        System.out.println("发送时间:"+localTime);
        channel.basicPublish(B_EX,B_ROUTING_KEY,null,msg.getBytes());
    }
    //测试延迟消息的消费者 打印消费时间 一致持续监听启动
    @Test
    public void consumeDelayMessage() throws IOException {
        channel.basicConsume(DL_QUEUE, true, (consumerTag, message) -> {
            //消费逻辑
            String msg=new String(message.getBody());
            LocalTime localTime = LocalDateTime.now().toLocalTime();
            System.out.println("消费消息:"+msg+";消费时间:"+localTime);
        }, consumerTag -> {});
        while(true);
    }
}
